package org.budo.support.service.batch.buffered;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

import org.budo.support.slf4j.Slf4j;
import org.slf4j.Logger;

/**
 * @author lmw
 */
public abstract class AbstractBufferedBatchService<T> {
    private static final Logger log = Slf4j.getLogger();

    private List<T> buffer = new ArrayList<T>();

    private Integer bufferSize = 20;

    private ExecutorService executorService;

    private Integer executorThreads = 2;

    private ExecutorService _executor;

    public void addToBuffer(T requestLog) {
        try {
            this.getBuffer().add(requestLog); // 加入缓冲

            if (this.getBuffer().size() >= this.getBufferSize()) { // 写入数据库
                this.submitFlashBuffer();
            }
        } catch (Throwable e) {
            log.error("#35 addToBuffer error, " + e, e);
        }
    }

    private void submitFlashBuffer() {
        final List<T> bufferCopy = new ArrayList<T>(buffer);
        this.getBuffer().clear();

        this.executor().submit(new Runnable() {
            public void run() {
                try {
                    AbstractBufferedBatchService.this.flashBuffer(bufferCopy);
                } catch (Throwable e) { // 捕获并日志异步方法里的异常
                    log.error("#48 _flashBuffer error, e=" + e, e);
                    throw e instanceof RuntimeException ? (RuntimeException) e : new RuntimeException(e);
                }
            }
        });
    }

    public ExecutorService executor() {
        if (null != this.getExecutorService()) {
            return this.getExecutorService();
        }

        if (null != this.getExecutorThreads()) {
            if (null == this._executor) {
                this._executor = Executors.newFixedThreadPool(this.getExecutorThreads());
            }

            return this._executor;
        }

        throw new RuntimeException("#68 executorService or executorThreads needs to be configured");
    }

    /**
     * 销毁前执行一遍
     */
    public void preDestroy() {
        this.flashBuffer(this.getBuffer());
        this.getBuffer().clear();
    }

    protected abstract void flashBuffer(List<T> bufferCopy);

    public Integer getExecutorThreads() {
        return this.executorThreads;
    }

    public void setExecutorThreads(Integer executorThreads) {
        this.executorThreads = executorThreads;
    }

    protected List<T> getBuffer() {
        return buffer;
    }

    public Integer getBufferSize() {
        return bufferSize;
    }

    public ExecutorService getExecutorService() {
        return executorService;
    }

    public void setBuffer(List<T> buffer) {
        this.buffer = buffer;
    }

    public void setBufferSize(Integer bufferSize) {
        this.bufferSize = bufferSize;
    }

    public void setExecutorService(ExecutorService executorService) {
        this.executorService = executorService;
    }
}
